package com.shujia.flink.core;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author SHUJIA
 */
public class Demo2Car {
    public static void main(String[] args) throws Exception {
        /*
         * 实时计算每隔城市每天总的车流量
         */

        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        DataStream<String> lines = env.socketTextStream("master", 8888);

        //解析数据取出需要的字段
        DataStream<Tuple3<String, String, Integer>> kvs = lines.map(line -> {
            //将json字符串转换成json对象
            JSONObject jsonObject = JSON.parseObject(line);
            //取出城市编码和时间
            String cityCode = jsonObject.getString("city_code");
            Long time = jsonObject.getLong("time");

            //将时间戳转换成时间字符串
            Date date = new Date(time * 1000);
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
            String day = format.format(date);

            //返回城市编码和日期
            return Tuple3.of(cityCode, day, 1);
        }, Types.TUPLE(Types.STRING, Types.STRING, Types.INT));

        //按照城市编号和日期分组，
        KeyedStream<Tuple3<String, String, Integer>, Tuple2<String, String>> keyBy = kvs
                .keyBy(kv -> Tuple2.of(kv.f0, kv.f1), Types.TUPLE(Types.STRING, Types.STRING));

        //统计车流量
        DataStream<Tuple3<String, String, Integer>> flow = keyBy.sum(2);

        flow.print();

        env.execute();
    }
}
